package com.intct.dwd;

import com.intct.common.FlinkSqlWithUtil;
import com.intct.func.DesensitizationFunction;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-23 09:37
 * 车辆分布功能
 */
public class DwdVehicleDistrSQL {
    public static void main(String[] args) {
        // 1 环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置全局并行度
        env.setParallelism(1);
        // 开启检查点，并且设置检查为一次语义
        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);
        // TableAPI环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 2 source映射 -- kafka
        // 样例：{"op":"r","name":"vehicle_distribution",
        // "after":{"actionTime":1758079820,"areaId":"120104","driverId":"363","carShelfNumber":"Y636160209D",
        // "car":"160","type":"0","speed":9},"ts_ms":1758079820459}
        tenv.executeSql("CREATE TABLE flink_vehicle (       " +
                "  op STRING                                         " +
                "  ,name STRING                                      " +
                "  ,after MAP<STRING, STRING>                        " +
                "  ,ts_ms BIGINT                                     " +
                "  ,pt AS PROCTIME()                                 " +
                ") WITH (                                            " +
                "  'connector'='kafka'                               " +
                "  ,'topic'='ods_vehicle'                            " +
                "  ,'properties.bootstrap.servers'='cdh-node:9092'   " +
                "  ,'properties.group.id'='vehicle_group'            " +
                "  ,'format'='json'                                  " +
                "  ,'scan.startup.mode'='earliest-offset'            " +
                ")                                                   ");

        // 3 处理数据
        // TODO 3.1 规范化

        // 3.2. 转换 + 去重 + 过滤 + 脱敏
        // 3.2.1 注册脱敏函数
        tenv.createTemporarySystemFunction("desens", DesensitizationFunction.class);

        // 样例：{"op":"r","name":"vehicle_distribution",
        // "after":{"actionTime":1758079820,"areaId":"120104","driverId":"363","carShelfNumber":"Y636160209D",
        // "car":"160","type":"0","speed":9},"ts_ms":1758079820459}
        tenv.executeSql("CREATE TEMPORARY VIEW view_vehicle AS                " +
                "  SELECT                                                              " +
                "    after['actionTime'] AS action_time                                " +
                "    ,after['areaId'] AS area_idarea_id                                " +
                "    ,after['carShelfNumber'] AS car_shelf_uumber                      " +
                "    ,desens(after['carShelfNumber'], 3, 3) AS dese_car_shelf_number   " +
                "    ,after['car'] as car                                              " +
                // 临时视图中不能使用处理时间函数
                "    ,pt                                                               " +
                " FROM flink_vehicle                                                   " +
                // 过滤不在线车辆
                " WHERE after['type'] = '0'                                            ");

        // 3.3 退维
        // 3.3.1 区域--映射维度表(异步)
        tenv.executeSql(" CREATE TABLE hbase_area_info(             " +
                "     id	STRING                                           " +
                "     ,info  ROW<                                            " +
                "        name   STRING                                       " +
                "     >                                                      " +
                " )WITH(                                                     " +
                "   'connector' = 'hbase-2.2'                              " +
                "   ,'zookeeper.quorum' = 'cdh-node:2181'                  " +
                "   ,'table-name' = 'dim:dim_area_info'                    " +
                // Lookup 异步开启方式
                "     ,'lookup.async' = 'true'                               " +
                " )                                                          " );

        // 4 sink映射 -- dwd kafka
        tenv.executeSql("CREATE TABLE kafka_sink_vehcle(            " +
                "  action_time               TIMESTAMP(3)                    " +
                "  ,area_idarea_id           STRING                          " +
                "  ,car_shelf_uumber         STRING                          " +
                "  ,dese_car_shelf_number    STRING                          " +
                "  ,car                      STRING                          " +
                "  ,area_name                STRING                          " +
                "  ,PRIMARY KEY (area_idarea_id, action_time) NOT ENFORCED   " +
                ")" + FlinkSqlWithUtil.getKafkaSinkWith("dwd_vehicle"));

        // 5 插入(DWD)kafka
        tenv.executeSql("INSERT INTO kafka_sink_vehcle              " +
                " SELECT                                                     " +
                "   TO_TIMESTAMP(TO_DATE(string1[, string2])(CAST(v.action_time AS BIGINT), 'yyyy-MM-dd HH:mm:ss')) " +
                "   , v.area_idarea_id                                       " +
                "   , v.car_shelf_uumber                                     " +
                "   , v.dese_car_shelf_number                                " +
                "   , v.car                                                  " +
                "   , a.info.name                                            " +
                " FROM view_vehicle v                                        " +
                " LEFT JOIN hbase_area_info FOR SYSTEM_TIME AS OF v.pt AS a  " +
                " ON v.area_idarea_id = a.id");
    }
}
